package com.zm.xiaohashu.note.biz.consumer;

import cn.hutool.json.JSONUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.zm.xiaohashu.note.biz.constant.MQConstants;
import com.zm.xiaohashu.note.biz.domain.NoteLike;
import com.zm.xiaohashu.note.biz.mapper.NoteLikeMapper;
import com.zm.xiaohashu.note.biz.model.dto.LikeUnlikeNoteMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Objects;

@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_LIKE_OR_UNLIKE, // Group 组
        topic = MQConstants.TOPIC_LIKE_OR_UNLIKE, // 消费的主题 Topic
        consumeMode = ConsumeMode.ORDERLY // 设置为顺序消费模式
)
@Slf4j
public class LikeUnlikeNoteConsumer implements RocketMQListener<Message> {

    // 每秒创建 5000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(5000);
    @Resource
    private NoteLikeMapper noteLikeMapper;
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();
        String bodyJsonStr = new String(message.getBody());
        // 标签
        String tags = message.getTags();

        log.info("==> LikeUnlikeNoteConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);
        // 根据 MQ 标签，判断操作类型
        if (Objects.equals(tags, MQConstants.TAG_LIKE)) { // 点赞笔记
            handleLikeNoteTagMessage(bodyJsonStr);
        } else if (Objects.equals(tags, MQConstants.TAG_UNLIKE)) { // 取消点赞笔记
            handleUnlikeNoteTagMessage(bodyJsonStr);
        }

    }

    private void handleLikeNoteTagMessage(String bodyJsonStr) {
        if (bodyJsonStr == null) return;
        LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO = JSONUtil.toBean(bodyJsonStr, LikeUnlikeNoteMqDTO.class);
        Integer type = likeUnlikeNoteMqDTO.getType();
        Long userId = likeUnlikeNoteMqDTO.getUserId();
        Long noteId = likeUnlikeNoteMqDTO.getNoteId();
        NoteLike noteLike = NoteLike.builder()
                .userId(userId)
                .noteId(noteId)
                .status(type)
                .build();
        noteLikeMapper.insertOrUpdateNoteLike(noteLike);
        sendMessage(bodyJsonStr);
    }


    private void handleUnlikeNoteTagMessage(String bodyJsonStr) {
        log.info("==> 取消点赞笔记");
        if (bodyJsonStr == null) return;
        LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO = JSONUtil.toBean(bodyJsonStr, LikeUnlikeNoteMqDTO.class);
        Integer type = likeUnlikeNoteMqDTO.getType();
        Long userId = likeUnlikeNoteMqDTO.getUserId();
        Long noteId = likeUnlikeNoteMqDTO.getNoteId();
        LambdaUpdateWrapper<NoteLike> luw = new LambdaUpdateWrapper<>();
        luw.eq(NoteLike::getUserId, userId)
                .eq(NoteLike::getNoteId, noteId)
                .set(NoteLike::getStatus, 0);
        int i = noteLikeMapper.update(null, luw);
        System.out.println(i);
        sendMessage(bodyJsonStr);
    }


    private void sendMessage(String bodyJsonStr) {
        //笔记计数
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();
        String noteDestination = MQConstants.TOPIC_NOTE_LIKE_OR_UNLIKE_COUNT;
        rocketMQTemplate.asyncSend(noteDestination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务 笔记点赞or取消点赞】MQ 推送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务 笔记点赞or取消点赞】MQ 推送异常: ", throwable);
            }
        });
        String userDestination = MQConstants.TOPIC_USER_LIKE_OR_UNLIKE_COUNT;
        rocketMQTemplate.asyncSend(userDestination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：用户点赞数入库】MQ 推送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：用户点赞数入库】MQ 推送异常: ", throwable);
            }
        });

    }

}
